package com.hzgj.bcl.soa.cicada;

import com.hzgj.bcl.soa.Event;
import com.hzgj.bcl.soa.cicada.lb.LoadBalancer;
import com.hzgj.bcl.soa.cicada.lb.Rule;
import com.hzgj.bcl.soa.cicada.sender.Sender;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.soa.util.Util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import lombok.AllArgsConstructor;

/**
 * Created by warning5 on 2017/3/3.
 */
public class Cicada {

    private Logger logger = LoggerFactory.getLogger(getClass());
    private LoadBalancer loadBalancer;
    private CicadaCallback cicadaCallback;

    public Cicada(LoadBalancer loadBalancer, CicadaCallback cicadaCallback) {
        this.loadBalancer = loadBalancer;
        this.cicadaCallback = cicadaCallback;
    }

    public void start() {
        loadBalancer.start();
    }

    public void stop() {
        loadBalancer.stop();
    }

    public Object send(Event evt) {
        Message message = (Message) evt.getArg();
        sendRequest(message);
        return null;
    }

    private void sendRequest(Message message) {
        try {
            Boolean isAsyn = (Boolean) message.getHeader(Constants.MESSAGE_HEADER_ASYN);
            if (isAsyn) {
                asyn(message);
            } else {
                syn(message);
            }
        } catch (Exception e) {
            logger.error("{}", e);
            handleThrowable(e, message.getHeaders());
        }
    }

    private void syn(final Message message) throws Exception {
        Sender sender = getSender(message);
        BasicAsyncHandler handler = new BasicAsyncHandler(message, sender) {
            @Override
            public void onThrowable(Throwable t) {
                try {
                    clusterHandle(message, sender, t);
                } catch (Throwable throwable) {
                    handleThrowable(t, message.getHeaders());
                }
            }

            @Override
            public Message onCompleted(Message message) throws Exception {
                return message;
            }
        };
        int requestTimeoutInMills = (int) message.getHeader(Constants.MESSAGE_HEADER_REQUESTTIMEOUTINMILLS);
        Object response = null;
        String service = message.getService();
        String method = message.getMethod();
        try {
            String poolKey = service + "_" + method;
            if (requestTimeoutInMills < 0) {
                response = sender.send(message, handler, requestTimeoutInMills, poolKey).get();
            } else {
                response = sender.send(message, handler, requestTimeoutInMills, poolKey).get(requestTimeoutInMills, TimeUnit.MILLISECONDS);
            }
        } catch (TimeoutException e) {
            logger.error("can't get response from " + service + "[" + method + "]" + " within " + requestTimeoutInMills + "ms");
            handleTimeout(e, message);
        } catch (ExecutionException e) {
            if (e.getCause() instanceof IOException) {
                if (!e.getCause().getMessage().equals(Util.REMOTELY_CLOSED_EXCEPTION.getMessage())) {
                    throw e;
                }
            }
        } catch (Exception e) {
            handler.onThrowable(e);
        }
        //up调用出错，不会触发 handler.onThrowable(e);
        //如果触发了 handler.onThrowable(e);response=null，不再触发回调
        if (response != null) {
            cicadaCallback.up(new Event(Event.MSG, response));
        }
    }

    private void asyn(final Message message) throws Exception {
        Sender sender = getSender(message);
        sender.send(message, new BasicAsyncHandler(message, sender), (int) message.getHeader(Constants.MESSAGE_HEADER_REQUESTTIMEOUTINMILLS),
                message.getService() + "_" + message.getMethod());
    }

    private Sender getSender(Message message) throws Exception {
        String service = message.getService();
        String method = message.getMethod();
        String rule = message.getLbRule();
        Sender sender = loadBalancer.chooseSender(Rule.RuleObject.builder().method(method).service(service).rule(rule).build());
        if (sender == null) {
            throw new Exception("no sender for service = " + service + ",method = " + method);
        }
        return sender;
    }

    void clusterHandle(Message message, Sender sender, Throwable t) throws Throwable {
        switch ((String) message.getHeader(Constants.MESSAGE_HEADER_CLUSTER)) {
            case Constants.DEFAULT_CLUSTER_FAILOVER:
                handleFailover(message, sender, t);
                break;
            case Constants.CLUSTER_FAILFAST:
                handleThrowable(t, message.getHeaders());
                break;
            case Constants.CLUSTER_FAILSAFE:
                logger.error("{}", t);
                throw t;
        }
    }

    private void handleFailover(Message message, Sender sender, Throwable t) throws Throwable {
        loadBalancer.markSenderDown(sender, (String) message.getHeader(Constants.MESSAGE_HEADER_TARGET_SERVICE));
        sendRequest(message);
    }

    private void handleTimeout(Throwable t, Message message) {
        message.putHeader(Constants.MESSAGE_HEADER_ATTACHMENT, t);
        cicadaCallback.up(new Event(Event.REINVOKE, message));
    }

    private void handleThrowable(Throwable t, Map<String, Object> headers) {
        Message message = Util.buildThrowableMessage(t, headers, (String) headers.get(Constants.MESSAGE_HEADER_PROTOCOL));
        message.putHeader(Constants.MESSAGE_HEADER_SERIALIZATION_BYPASS, true)
                .putHeader(Constants.MESSAGE_TYPE, Message.EXC_RSP);
        cicadaCallback.up(new Event(Event.MSG, message));
    }

    public interface CicadaCallback {
        void up(Event event);
    }

    @AllArgsConstructor
    class BasicAsyncHandler extends AsyncCompletionHandlerBase<Message> {
        // 发送的消息
        Message send_message;
        Sender cicadaSender;

        @Override
        public void onThrowable(Throwable t) {
            if (t instanceof TimeoutException) {
                handleTimeout(t, send_message);
            } else {
                try {
                    clusterHandle(send_message, cicadaSender, t);
                } catch (Throwable throwable) {
                    handleThrowable(t, send_message.getHeaders());
                }

            }
        }

        @Override
        public Message onCompleted(Message message) throws Exception {
            cicadaCallback.up(new Event(Event.MSG, message));
            return message;
        }
    }
}
